package com.hou.elasticsearch.config;

import org.elasticsearch.action.bulk.BackoffPolicy;
import org.elasticsearch.action.bulk.BulkProcessor;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.unit.TimeValue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * 批量操作数据对象配置
 */
@Configuration
public class BulkProcessorConfig {

    @Autowired
    private TransportClient transportClient;

    private static TransportClient client;

    /**
     * @PostContruct 是spring容器初始化的时候执行该方法
     */
    @PostConstruct
    public void init() {
        client = this.transportClient;
    }

    /**
     * 在需要保证数据的准确性的场景下，rest方式并不能保证结果的准确性，因此采用了elasticsearch的BulkProcessor方式来进行数据入库，
     * 实际上采用es客户端不同，rest方式采用的是restClient，基于http协议，BulkProcessor使用的是TransportClient，基于Tcp协议
     *
     * @return
     */
    //配置BulkProcessor对象用来批量操作数据
    @Bean("bulkProcessor")
    public BulkProcessor bulkProcessor() {
        return BulkProcessor.builder(client, new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long l, BulkRequest bulkRequest) {
            }

            @Override
            public void afterBulk(long l, BulkRequest bulkRequest, BulkResponse bulkResponse) {
            }

            @Override
            public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
            }
        }).setBulkActions(10000)  // 批量导入个数
                .setBulkSize(new ByteSizeValue(5, ByteSizeUnit.MB))  // 满5MB进行导入
                .setFlushInterval(TimeValue.timeValueSeconds(5))  // 冲刷间隔
                .setConcurrentRequests(3)  // 并发数
                .setBackoffPolicy(BackoffPolicy.exponentialBackoff(TimeValue.timeValueSeconds(1), 3))  //重试3次，间隔1s
                .build();
    }
}
